package app.dim;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import utils.MyKafkaUtil;
import utils.MyPhoenixSink;

import java.util.Random;

/**
 * 品牌维度表处理器
 */
public class DimBrandProcessor {
    
    public static void main(String[] args) throws Exception {
        // 创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        // 从Kafka读取商品数据，提取品牌信息
        String topic = "ods_market_product";
        String groupId = "dim_brand_group";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        
        // 转换数据格式，提取品牌信息
        kafkaDS.map(jsonStr -> {
            JSONObject jsonObject = JSONObject.parseObject(jsonStr);
            JSONObject dimJson = new JSONObject();
            dimJson.put("sinkTable", "dim_brand_info");
            
            JSONObject data = new JSONObject();
            String brandId = jsonObject.getString("brand_id");
            data.put("id", brandId);
            data.put("brand_name", "品牌" + brandId.substring(1));
            
            // 随机生成一些品牌属性
            Random random = new Random(brandId.hashCode());
            data.put("country", getRandomCountry(random));
            data.put("establish_year", 1990 + random.nextInt(30));
            data.put("create_time", jsonObject.getString("ts"));
            
            dimJson.put("data", data);
            return dimJson;
        }).addSink(new MyPhoenixSink());
        
        // 执行任务
        env.execute("Brand Dimension Processing Job");
    }
    
    private static String getRandomCountry(Random random) {
        String[] countries = {"中国", "美国", "日本", "韩国", "德国", "法国", "英国", "意大利"};
        return countries[random.nextInt(countries.length)];
    }
}